package com.jan.activemq01.test;

import java.util.concurrent.TimeUnit;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerFactory;
import org.apache.activemq.broker.BrokerService;
import org.junit.Test;
//生产者
public class ActiveMqTest01 {
	@Test
	public void test01() throws JMSException, InterruptedException{
		//ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("user", "123456", ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://106.15.92.48:61616");
		Connection con = factory.createConnection();
		con.start();
		Session session = con.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
		Queue que = session.createQueue("bbbb");
		MessageProducer producer = session.createProducer(null);
		
		for(int i=0;i<10;i++){
			TextMessage message = session.createTextMessage("我是消息"+i);
			message.setStringProperty("name", "name1");
			producer.send(que,message);
			TimeUnit.SECONDS.sleep(2);
		}
		session.commit();
		session.close();
		if(con!=null){
			con.close();
		}
	}
	
	@Test
	public void test2() throws JMSException, InterruptedException{
		//主题类发送 非持久topic必须客户端在线才能发送消息
		//ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("user", "123456", ActiveMQConnectionFactory.DEFAULT_BROKER_URL);
		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://106.15.92.48:61616");
		Connection con = factory.createConnection();
		con.start();
		Session session = con.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
		Topic topic = session.createTopic("top1");
		MessageProducer producer = session.createProducer(null);
		
		//设置持久化topic
		//producer.setDeliveryMode(DeliveryMode.PERSISTENT);
		//con.start();
		for(int i=0;i<10;i++){
			TextMessage message = session.createTextMessage("我是消息"+i);
			producer.send(topic,message);
			TimeUnit.SECONDS.sleep(2);
		}
		if(con!=null){
			con.close();
		}
	}
	
	@Test
	public void test3() throws Exception{
		BrokerService service = new BrokerService();
		service.setUseJmx(true);
		service.addConnector("tcp://localhost:61616");
		service.start();
	}
	
	@Test
	public void test4() throws JMSException, InterruptedException{
		//发送持久化topic
		//ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("user", "123456", "tcp://localhost:61616");
		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://106.15.92.48:61616");
		Connection con = factory.createConnection();
		Session session = con.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
		Topic topic = session.createTopic("top1");
		MessageProducer producer = session.createProducer(null);
		producer.setDeliveryMode(DeliveryMode.PERSISTENT);
		con.start();
		
		for(int i=0;i<10;i++){
			TextMessage message = session.createTextMessage("我是持久化消息"+i);
			producer.send(topic,message);
		}
		if(con!=null){
			con.close();
		}
	}
	@Test
	public void createBrokerTest() throws Exception{
		String uri="properties:broker.properties";
		//BrokerService brokerService = new BrokerService();
		BrokerService brokerService=BrokerFactory.createBroker(uri);
		brokerService.setUseJmx(true);
		brokerService.addConnector("tcp://127.0.0.1:6666");
		brokerService.start();
	}
}
